﻿// ———————————————————————–
// <copyright file="ContentFeedProcessing.cs" company="The MITRE Corporation">
//    Copyright (c) 2014 The MITRE Corporation. All rights reserved.
//    Licensed under the Apache License, Version 2.0 (the "License");
//    you may not use this file except in compliance with the License.
//    You may obtain a copy of the License at
//    http://www.apache.org/licenses/LICENSE-2.0
//    Unless required by applicable law or agreed to in writing, software
//    distributed under the License is distributed on an "AS IS" BASIS,
//    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//    See the License for the specific language governing permissions and
//    limitations under the License.
// </copyright>
// ———————————————————————–
/////////////////////////////////////////////////////////////////////////////////////////////////
// DEProcessing.cs - Service implementation that pulls DE messages from a SQS queue and writes
//                   them to DynamoDB for content feed purposes
// Project: EDXLSharp_AWSRouter - ContentFeedProcessing
//
// Language:    C#, .NET 4.0
// Platform:    Windows Server 2012, VS 2013
// Author:      Brian Wilkins The MITRE Corporation
/////////////////////////////////////////////////////////////////////////////////////////////////

// Copyright (c) 2014 The MITRE Corporation. All rights reserved.
//
// NOTICE
// This software was produced for the U. S. Government
// under Contract No. FA8721-09-C-0001, and is
// subject to the Rights in Noncommercial Computer Software
// and Noncommercial Computer Software Documentation Clause
// (DFARS) 252.227-7014 (JUN 1995)

using Amazon.DynamoDBv2;
using Amazon.DynamoDBv2.Model;
using Amazon.SQS;
using Amazon.SQS.Model;
using Amazon.Util;
using EDXLSharp.EDXLDELib;
using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Diagnostics;
using System.Linq;
using System.ServiceProcess;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using ICNETServices;
using EDXLSharp;
using ICNETDAL;
// Configure log4net using the .config file
[assembly: log4net.Config.XmlConfigurator(Watch = true)]
namespace ContentFeedProcessing
{
  /// <summary>
  /// Class for the Content Feed Processing Windows Service
  /// </summary>
  public partial class ContentFeedProcessing : ServiceBase
  {
    /// <summary>
    /// Log4net logging object
    /// </summary>
    private static readonly log4net.ILog Log = log4net.LogManager.GetLogger(System.Reflection.MethodBase.GetCurrentMethod().DeclaringType);

    /// <summary>
    /// Thread to run the SQS polling on
    /// </summary>
    private Thread runThread;

    /// <summary>
    /// Amazon SQS configuration object 
    /// </summary>
    private AmazonSQSConfig amazonSQSConfig;

    /// <summary>
    /// Amazon SQS Client
    /// </summary>
    private AmazonSQSClient amazonSQSClient;

    /// <summary>
    /// Amazon DynamoDB configuration object
    /// </summary>
    private AmazonDynamoDBConfig amazonDynamoConfig;

    /// <summary>
    /// Amazon DynamoDB client
    /// </summary>
    private AmazonDynamoDBClient amazonDynamoClient;


    /// <summary>
    /// Initializes a new instance of the ContentFeedProcessing class
    /// </summary>
    public ContentFeedProcessing()
    {
      this.InitializeComponent();
      this.ServiceName = "Content Feed Service";
    }

    /// <summary>
    /// Method called when the service is started
    /// </summary>
    /// <param name="args">Command arguments (ignored)</param>
    protected override void OnStart(string[] args)
    {
      Log.Info("OnStart");
      this.amazonSQSConfig = new AmazonSQSConfig();
      this.amazonSQSConfig.ServiceURL = AWSConstants.SQSServiceURL;
      this.amazonSQSClient = new AmazonSQSClient(this.amazonSQSConfig);
      this.amazonDynamoConfig = new AmazonDynamoDBConfig();
      this.amazonDynamoConfig.ServiceURL = AWSConstants.DynamoServiceURL;
      this.amazonDynamoClient = new AmazonDynamoDBClient(this.amazonDynamoConfig);
      this.runThread = new Thread(new ThreadStart(this.PollSQS));
      this.runThread.IsBackground = true;
      this.runThread.Start();
      Log.Info("OnStartComplete");
    }

    protected override void OnStop()
    {
      Log.Info("OnStop");
      if (this.runThread.IsAlive)
      {
        Log.Info("KillingThread");
        this.runThread.Abort();
        Log.Info("ThreadKilled");
      }
    }

    /// <summary>
    /// Thread worker method.
    /// </summary>
    private void PollSQS()
    {
      Log.Info("InThread");
      while (true)
      {
        Log.Info("SQSLoop");
        ReceiveMessageRequest req = new ReceiveMessageRequest();
        req.QueueUrl = AWSConstants.ContentFeedProcessingQueueUrl;
        req.WaitTimeSeconds = AWSConstants.ContentFeedProcessing.PollIntervalInSeconds;

        ReceiveMessageResponse resp = this.amazonSQSClient.ReceiveMessage(req);
        Log.Info("Got " + resp.Messages.Count + " messages from SQS");

        foreach (Message m in resp.Messages)
        {
          Log.Info(m.Body);
          EDXLDE de = new EDXLDE(m.Body);

          //assumes de clean process will delete data
          if (de.DistributionType != TypeValue.Cancel)
          {
            this.AddContentFeeds(de, m.Body);
          }

          DeleteMessageRequest deleteMessageRequest = new DeleteMessageRequest();
          deleteMessageRequest.QueueUrl = req.QueueUrl;
          deleteMessageRequest.ReceiptHandle = m.ReceiptHandle;
          try
          {
            DeleteMessageResponse sqsdelresp = this.amazonSQSClient.DeleteMessage(deleteMessageRequest);
            if (sqsdelresp.HttpStatusCode == System.Net.HttpStatusCode.OK)
            {
              Log.Info("Message removed from queue");
            }
            else
            {
              Log.Error("SQS Delete error: " + sqsdelresp.HttpStatusCode.ToString());
            }
          }
          catch (Exception e)
          {
            Log.Error("Error deleting from SQS", e);
          }
        }
      }
    }

    /// <summary>
    /// Adds the DE content object feeds relationship to the FeedCache table.
    /// </summary>
    /// <param name="de">The Distribution Element object containing the content objects.</param>
    /// <param name="body">A string of the message body.</param>
    private void AddContentFeeds(EDXLDE de, string body)
    {
      foreach (ContentObject co in de.ContentObjects)
      {
        //for now just use the ContentKeywords
        //for each content keyword get a rule
        foreach(ValueList vl in co.ContentKeyword)
        {
          //for each value in the value list
          foreach (string val in vl.Value)
          {
            //get the list of feed hashes that are associated to this valuelist urn and value
            Dictionary<string, AttributeValue> feedHashes = GetFeedHashes(vl.ValueListURN, val);

            //should only be 1 attribute returned if there was a rule match
            if (feedHashes != null && feedHashes.Count == 1)
            {
              //add each hash in the string set
              foreach (string feedHash in feedHashes.Values.First().SS)
              {
                PutItemRequest putRequest = new PutItemRequest();
                putRequest.TableName = AWSConstants.ContentFeedProcessing.DynamoInfo.TableName;
                Dictionary<string, AttributeValue> dynamoItem = new Dictionary<string, AttributeValue>();

                //add contentfeeds hash key
                AttributeValue putVal = new AttributeValue();
                putVal.S = feedHash;
                dynamoItem.Add(AWSConstants.ContentFeedProcessing.DynamoInfo.HashName, putVal);

                //add contentfeeds range key
                putVal = new AttributeValue();
                putVal.N = DEDalUtils.ComputeHash(de.DistributionID, de.SenderID, co.ContentDescription).ToString();
                dynamoItem.Add(AWSConstants.ContentFeedProcessing.DynamoInfo.RangeName, putVal);

                //add content _ExpiresTime
                dynamoItem.Add("_ExpiresTime", new AttributeValue() {
                  S = this.GetExpiresTime(de, co)
                });

                putRequest.Item = dynamoItem;

                Log.Debug("Adding to ContentFeed Table - " + AWSUtils.CreateLogMessage(dynamoItem));
                try
                {
                  PutItemResponse dynamoResp = this.amazonDynamoClient.PutItem(putRequest);
                  if (dynamoResp.HttpStatusCode == System.Net.HttpStatusCode.OK)
                  {
                    Log.Info("Message Added to Dynamo ContentFeed Table");
                  }
                  else
                  {
                    Log.Error("Dynamo Error" + dynamoResp.HttpStatusCode.ToString());
                  }
                }
                catch (Exception e) //complete break out of the loop
                {
                  Log.Error("Error writing to DE to Dynamo ContentFeed Table", e);
                  return;
                }
              }
            }
            else if (feedHashes != null && feedHashes.Count > 1)
            {
              Log.Error("Unexpected number of Rules returned from GetFeedHashes (> 1).  Should be 1, 0, or null.");
            }
          }
        }
      }
    }

    private Dictionary<string, AttributeValue> GetFeedHashes(string ValListUrn, string ValListValue)
    {
      GetItemRequest request = new GetItemRequest();
      request.TableName = AWSConstants.RulesProcessing.DynamoInfo.TableName;
      request.ProjectionExpression = "#fh";
      request.ExpressionAttributeNames.Add("#fh", "_FeedHash");
      //add hashkey
      AttributeValue hashvalue = new AttributeValue();
      hashvalue.S = ValListUrn; 
      request.Key.Add(AWSConstants.RulesProcessing.DynamoInfo.HashName, hashvalue);

      //add range
      hashvalue = new AttributeValue();
      hashvalue.S = ValListValue;
      request.Key.Add(AWSConstants.RulesProcessing.DynamoInfo.RangeName, hashvalue);

      try
      {
        GetItemResponse response = this.amazonDynamoClient.GetItem(request);

        return response.Item;        
      }
      catch (Exception e)
      {
        Log.Error("Error querying from dynamo for latest", e);
        return null;
      }
    }

    private string GetExpiresTime(EDXLDE DE, ContentObject CO)
    {
      DateTime coExpires;

      if (CO.ExpiresTime.HasValue)
      {
        coExpires = CO.ExpiresTime.Value;
      }
      else
      {
        coExpires = DE.DateTimeSent.AddDays(1.0);
      }

      return coExpires.ToUniversalTime().ToString(AWSSDKUtils.ISO8601DateFormat);
    }
  }
}
